package org.devgframwork.datasync.core.support.init.rabbitmq;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.devgframwork.datasync.core.entity.TriggerConfig;
import org.devgframwork.datasync.core.model.ExecutorModel;
import org.devgframwork.datasync.core.model.ResultModel;
import org.devgframwork.datasync.core.support.DataSourceMemory;
import org.devgframwork.datasync.core.support.executors.AbstractProxyExecutor;
import org.devgframwork.datasync.core.utils.CommonUtil;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description
 * @auther dh
 * @date 2022/1/5 18:55
 */
@Slf4j
public class RabbitQueueConsumerInit {

    private static final Integer consumerThreadNumber = 10;

    private static final Integer retryInterval = 5000;

    private static final String retryCountStr = "retryCount";

    private TriggerConfig triggerConfig;

    public RabbitQueueConsumerInit(TriggerConfig triggerConfig) {
        this.triggerConfig = triggerConfig;
    }

    public static RabbitQueueConsumerInit getInstance(TriggerConfig triggerConfig) {
        return new RabbitQueueConsumerInit(triggerConfig);
    }

    public void initConsumer() throws IOException {
        String normal = triggerConfig.getNormalQueue();
        String retry = triggerConfig.getRetryQueue();
        String error = triggerConfig.getErrorQueue();
        String exchange = triggerConfig.getExchange();
        String connectionName = triggerConfig.getConnectionName();
        Channel channel = DataSourceMemory.connectionChannelMap.get(connectionName);
//        channel.queueDeclare(normal, true, false, false, null);
        log.info("Customer queue:{} Waiting Received messages", normal);
        //DefaultConsumer类实现了Consumer接口，通过传入一个频道，
        // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
        Consumer consumer = new DefaultConsumer(channel) {
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, CommonUtil.charset);
//                Integer expiration = 5000;
//                AMQP.BasicProperties reProperties = new AMQP.BasicProperties.Builder().expiration(expiration.toString()).build();
//                channel.basicPublish("datax-node-jdbc-error-exchange", "re_wait_"+QUEUE_NAME, reProperties, message.getBytes("utf-8"));
                log.info("Customer Received '" + message + "'");
                String routingKey = envelope.getRoutingKey();
                String contentType = properties.getContentType();
                long deliveryTag = envelope.getDeliveryTag();
                try {
                    ExecutorModel executorModel = new ExecutorModel(routingKey, triggerConfig.getExecuteName());
                    executorModel.setExt(message);
                    AbstractProxyExecutor abstractProxyExecutor = new AbstractProxyExecutor(executorModel);
                    ResultModel resultModel = abstractProxyExecutor.invoke();
                    log.info("routingKey:{},execute resultModel:{} ", routingKey, JSONObject.toJSONString(resultModel));
                } catch (Exception ex) {
                    //这里是模拟出错情况,先写入重试队列,5秒后进入消费队列
                    int retryCount = getRetryCount(properties);
                    log.error("失败次数：" + retryCount);
                    if (retryCount >= 3) {
                        channel.queueDeclare(error, true, false, false, null);
                        channel.queueBind(error, exchange, error);
                        channel.basicPublish(exchange, error, null, message.getBytes(CommonUtil.charset));
                        channel.basicAck(deliveryTag, false);
                        return;
                    }
                    Map<String, Object> headers = new HashMap<String, Object>();
                    headers.put(retryCountStr, retryCount + 1);
                    String expiration = getExpiration(retryCount + 1);
                    AMQP.BasicProperties reProperties = new AMQP.BasicProperties.Builder().expiration(expiration).headers(headers).build();
                    channel.basicPublish(exchange, retry, reProperties, message.getBytes(CommonUtil.charset));
                    channel.basicAck(deliveryTag, false);
                    return;
                }
                channel.basicAck(deliveryTag, false);
            }
        };
        //自动回复队列应答 -- RabbitMQ中的消息确认机制
//  channel.basicConsume(QUEUE_NAME, true, consumer);
        channel.basicQos(5);
        channel.basicConsume(normal, false, consumer);//这里改为手动确认
    }

    public int getRetryCount(AMQP.BasicProperties properties) {
        Integer retryCount = 0;
        Map<String, Object> headers = properties.getHeaders();
        if (headers != null && headers.containsKey(retryCountStr)) {
            retryCount = (Integer) headers.get(retryCountStr);
        }
        return retryCount;
    }

    public String getExpiration(int retryCount) {
        return String.valueOf(retryInterval * retryCount);
    }

}
